package org.example.jasper
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions._

object BZLocatDiff {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("BZ")
    val sc = new SparkContext(conf)

    // 创建 SparkSession
    val spark = SparkSession.builder
      .appName("BZLocatDiff")
      .getOrCreate()

    // 读取数据
    val inputPath = "src/main/java/org/datas/bz"
    val df = spark.read
      .option("header", "true")
      .csv(inputPath)

    // 使用groupBy和agg统计每个发生地点的车祸次数
    val result = df
      .groupBy("uso_solo")
      .agg(count("*").alias("accident_count"))
      .orderBy(desc("accident_count"))

    // 打印结果
    result.show()

    // 将结果导入到 MySQL 数据库中
    val mysqlUrl = "jdbc:mysql://localhost:3306/BZaccident"
    val mysqlProperties = new java.util.Properties()
    mysqlProperties.setProperty("user", "root")
    mysqlProperties.setProperty("password", "011216")
    mysqlProperties.setProperty("driver", "com.mysql.jdbc.Driver")

    result.write
      .mode(SaveMode.Overwrite) // 如果表已存在，覆盖之前的数据
      .jdbc(mysqlUrl, "BZLocatDiff", mysqlProperties)
  }
}
